package com.example.rabbitextensionsdemo.controller;

import com.example.rabbitextensionsdemo.constant.Constants;
import com.rabbitmq.client.BasicProperties;
import com.rabbitmq.client.Channel;

import jakarta.annotation.Resource;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.Date;

@RestController
@RequestMapping("/rabbit")
public class ProducerController {
    @Resource(name = "rabbitTemplate")
    private RabbitTemplate rabbitTemplate = new RabbitTemplate();
    @Resource(name = "confirmRabbitTemplate")
    private RabbitTemplate confirmRabbitTemplate = new RabbitTemplate();
    @Resource(name = "transRabbitTemplate")
    RabbitTemplate transRabbitTemplate = new RabbitTemplate();
    // 自动确认
    @RequestMapping("/ack")
    public String ack() {
        rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,"ack","consumer ack text...");
        return "发送成功";
    }
    // 持久化
    // 消息不持久
    @RequestMapping("/pres")
    public String pres() throws IOException {
        Message message = new Message("pres  text...".getBytes(),new MessageProperties());
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 消息持久化
//        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); // 消息非持久化
        rabbitTemplate.convertAndSend(Constants.PRES_EXCHANGE,"pres", message);
        return "发送成功";
    }
    // 发送确认
    @RequestMapping("/confirm")
    public String confirm() {
        String message = "confirm text...";
        CorrelationData correlationData = new CorrelationData("1");
        confirmRabbitTemplate.convertAndSend(Constants.CONFIRM_EXCHANGE,"confirm111", message.getBytes(),correlationData);
        return "发送成功";
    }
    @RequestMapping("/returns")
    public String returns() {
        String message = "returns test...";
        CorrelationData correlationData = new CorrelationData();
        confirmRabbitTemplate.convertAndSend(Constants.CONFIRM_EXCHANGE, "confirm",message.getBytes(),correlationData);
        return "发送成功";
    }
    // 重试机制
    @RequestMapping("/retry")
    public String retry() {
        String message = "retry test...";
        rabbitTemplate.convertAndSend(Constants.RETRY_EXCHANGE,"retry",message.getBytes());
        return "发送成功";
    }
    // TTL 过期时间
    // 队列和消息同时设置过期时间时, 以时间短的为基准
    @RequestMapping("/ttl")
    public String ttl() {
        String ttlTime = "20000"; // 二十秒钟
        // 消息过期
        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, "ttl", "ttl test...", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration(ttlTime);
                return message;
            }
        });
        return "发送成功";
    }
    // ttl存在的问题
    @RequestMapping("/ttl2")
    public String ttl2() {
        // 消息过期
        // 如果后一条消息的时间比队首的消息短, 后一条消息会等待队首过期后,才会进行丢弃
        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, "ttl", "ttl test is 30s...", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("30000");
                return message;
            }
        });
        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, "ttl", "ttl test is 20s...", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("20000");
                return message;
            }
        });
        return "发送成功";
    }
    @RequestMapping("/normal")
    public String normal() {
        // 测试队列溢出, 进入死信队列, 溢出10条
//        for(int i = 0; i < 20; i++) {
//            rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","dlx normal test...");
//        }
        // 测试消息超时
//        rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE, "normal", "dlx normal test...", new MessagePostProcessor() {
//            @Override
//            public Message postProcessMessage(Message message) throws AmqpException {
//                message.getMessageProperties().setExpiration("10000");
//                return message;
//            }
//        });
        // 测试消息拒收
        rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","dlx normal test...");
        return "发送成功";
    }
    // TTL+死信 实现延迟
    @RequestMapping("/delay")
    public String delay() {
        rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay normal test...");
        System.out.printf("发送消息 %tc\n", new Date());
        return "发送成功";
    }
    // 延时队列插件
    @RequestMapping("/delayed")
    public String delayed() {
        System.out.println("开始发送");
        rabbitTemplate.convertAndSend(Constants.DELAYED_EXCHANGE, "delayed", "delayed test 20S..." + new Date(), new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelayLong(20000L);
                return message;
            }
        });
        rabbitTemplate.convertAndSend(Constants.DELAYED_EXCHANGE, "delayed", "delayed test 30S..." + new Date(), new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelayLong(30000L);
                return message;
            }
        });
        System.out.printf("发送消息成功: %tc\n", new Date());
        return "发送成功";
    }
    @Transactional
    @RequestMapping("/trans")
    public String trans() {
        System.out.println("trans test...");
        transRabbitTemplate.convertAndSend("", "trans.queue", "trans test1...");
        int m = 5/0;
        transRabbitTemplate.convertAndSend("", "trans.queue", "trans test2...");
        return "发送成功";
    }
    // 限流
    @RequestMapping("/qos")
    public String qos() {

        for(int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE,"qos", "qos test " + i);
        }
        return "发送成功";
    }
}
